package com.example.toy.core.record.key.generator;

import com.example.toy.core.cons.ToyConst;
import com.example.toy.core.utils.SchemaHelper;
import com.example.toy.core.utils.StreamUtils;
import lombok.Getter;
import org.apache.avro.Schema;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * CombineKeyGenerator
 *
 * @author shenbing
 * @date 2021/12/17
 */
@Getter
public class CombineKeyGenerator implements KeyGenerator {

  @Override
  public String generate(Schema schema, Object... params) {
    if (params == null || params.length == 0) {
      throw new IllegalArgumentException("HashKeyGenerator params can not be null or empty.");
    }

    if (schema.getFields().size() != params.length) {
      throw new IllegalArgumentException("Key params length not equals the schema field's size.");
    }

    List<Schema.Field> fields = schema.getFields();
    String generated =
        StreamUtils.zip(
                fields.stream(),
                Arrays.stream(params),
                (field, param) -> {
                  return field.name() + "=" + String.valueOf(param);
                })
            .collect(Collectors.joining(","));
    return generated;
  }

  @Override
  public List<Object> fromKey(Schema schema, String key) {
    if (key.equals(ToyConst.NON_PARTITION_KEY)) {
      return new ArrayList<>();
    }
    String[] split = key.split(",");
    List<Object> objects = new ArrayList<>();
    if (schema.getFields().size() != split.length) {
      throw new IllegalArgumentException("Key params length not equals the schema field's size.");
    }
    for (int i = 0; i < split.length; i++) {
      String text = split[i];
      String value = text.split("=")[1];
      objects.add(SchemaHelper.convert(value, schema.getFields().get(i).schema()));
    }
    return objects;
  }

  public static void main(String[] args) {
    Schema user = Schema.createRecord("user", "", "", false);
    user.setFields(
        Arrays.asList(
            new Schema.Field("id", Schema.create(Schema.Type.STRING)),
            new Schema.Field("name", Schema.create(Schema.Type.INT))));

    CombineKeyGenerator generator = new CombineKeyGenerator();
    String generate = generator.generate(user, "xx", "1");
    System.out.println(generate);

    System.out.println(generator.fromKey(user, generate));
  }
}
